package org.flink.marketing.sink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.avro.data.Json;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.flink.marketing.constant.EventTrackingConstant;
import org.flink.marketing.constant.SqlConstant;
import redis.clients.jedis.Jedis;
import java.io.IOException;
import java.io.InputStream;
import java.sql.*;
import java.util.*;

@Slf4j
public class ClickHouseSinkUsers extends RichSinkFunction<JSONObject> {
    private Connection connection;
    private PreparedStatement preparedStatement;
    private transient Jedis jedis;
    private static Properties properties;

    private static void init() {
        properties = new Properties();
        try(InputStream inputStream = ClickHouseSinkUsers.class.getResourceAsStream("/ck.properties")) {
            properties.load(inputStream);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    @Override
    public void open(Configuration parameters) throws Exception {
        init();
        super.open(parameters);
        // 加载JDBC驱动
        Class.forName(properties.getProperty("DriverClass"));
        // 获取数据库连接
        connection = DriverManager.getConnection(properties.getProperty("DriverUrl"),properties.getProperty("User"),properties.getProperty("Password"));

    }

    @Override
    public void close() throws Exception {
        super.close();
        connection.close();
    }

    public static void initUser(JSONObject jsonValue) throws SQLException {

        Iterator<String> it = jsonValue.keySet().iterator();
        //处理用户埋点属性分支
        //用户埋点数据报文格式
        /*{
                "login_id": "123456789",
                "event_Name": "login_user_data",
                "properties": {
                    "sex": "男",
                    "age": "30",
                    "name": "张三"
                 }
        }*/
        //login_user_data 标识用户属性的事件
        //用户属性只采集UI登录成功之后的数据，用户ID为唯一表示，对用户表数据进行增加，删除，修改

        //第一步，查询用户数据是否存在
        String userId="";
        while(it.hasNext()){
            // 获得key
            String key = it.next();
            if(SqlConstant.LOGINID.equals(key)){
                userId = jsonValue.getString(key);
                break;
            }
        }

        //用户ID没有不处理
        if("".equals(userId)){
            return;
        }

        StringBuffer selectSql = new StringBuffer();
        selectSql.append(SqlConstant.SELECCTSQL).append(SqlConstant.WHERELOINGID).append(userId);
        String sql = selectSql.toString();
        log.info("执行用户查询操作....................:"+sql);

        /*PreparedStatement preparedStatement = connection.prepareStatement(sql);
        ResultSet resultSet = preparedStatement.executeQuery();
*/
        List<String> tableColums = new ArrayList<>();
        List<String> fileds = new ArrayList<>();

        //更新
        if(false){

            //properties 节点属性
            JSONObject propertiesJsonValue = jsonValue.getJSONObject(EventTrackingConstant.PROPERTIES);
            Iterator<String> itProperties = propertiesJsonValue.keySet().iterator();

            StringBuffer s=new StringBuffer();
            while(itProperties.hasNext()){
                // 获得key
                String key = itProperties.next();
                String value = propertiesJsonValue.getString(key);
                value = value == null?"":value;
                tableColums.add(value);
                s.append(key).append(" = ").append(" ? ").append(",");
            }

            StringBuffer updateSql = new StringBuffer();
            updateSql.append(SqlConstant.UPDATESQL);

            updateSql.append(s.substring(0,s.length()-1)).append(SqlConstant.WHERELOINGID).append(userId);

            String isql = updateSql.toString();
            log.info("用户数据更新SQL.......:"+isql);

         /*  *//* PreparedStatement updateConnection = connection.prepareStatement(isql);

            for (int i = 0; i < tableColums.size(); i++) {
                updateConnection.setString(i + 1, tableColums.get(i));
            }

            int ints = updateConnection.executeUpdate();
            long startTime = System.currentTimeMillis();
            connection.commit();*//*
            long endTime = System.currentTimeMillis();
            log.info("批量更新用户数据完毕用时：" + (endTime - startTime) + " -- 插入数据 = " + ints);*/

        }

        //新增
        else{

            StringBuffer insertSql = new StringBuffer();
            insertSql.append(SqlConstant.INSERTSQL);
            insertSql.append("(");

            insertSql.append(SqlConstant.LOGINID).append(",");
            fileds.add(SqlConstant.LOGINID);
            tableColums.add(userId);

            StringBuffer tsql=new StringBuffer();

            //properties 节点属性
            JSONObject propertiesJsonValue = jsonValue.getJSONObject(EventTrackingConstant.PROPERTIES);
            Iterator<String> itProperties = propertiesJsonValue.keySet().iterator();
            while(itProperties.hasNext()){
                // 获得key
                String key = itProperties.next();
                String value = propertiesJsonValue.getString(key);
                if (fileds.contains(key)){
                    continue;
                }

                tsql.append(key).append(",");
                value = value == null?"":value;
                tableColums.add(value);
                fileds.add(key);
            }

            insertSql.append(tsql.substring(0,tsql.length()-1)).append(")");
            insertSql.append(SqlConstant.INSERT_VALUE).append("(");
            int length = tableColums.size();
            int t=0;
            for (String s :tableColums){
                if (t+1 == length){
                    insertSql.append("?");
                }else {
                    insertSql.append("?").append(",");
                }
                t++;
            }
            insertSql.append(")");

            String isql = insertSql.toString();

            log.info("用户数据新增SQL.......:"+isql);

          /*  PreparedStatement insertConnection = connection.prepareStatement(isql);

            for (int i = 0; i < length; i++) {
                insertConnection.setString(i + 1, tableColums.get(i));
            }

            int ints = insertConnection.executeUpdate();
            long startTime = System.currentTimeMillis();
            connection.commit();
            long endTime = System.currentTimeMillis();
            log.info("批量插入用户用户完毕用时：" + (endTime - startTime) + " -- 插入数据 = " + ints);*/
        }
    }


    public static void main(String args[]) throws SQLException {
        System.out.println("Hello World!");

        String s="{\n" +
                "                \"userId\": \"123456789\",\n" +
                "                \"eventName\": \"login_user_data\",\n" +
                "                \"properties\": {\n" +
                "                    \"sex\": \"男\",\n" +
                "                    \"age\": \"30\",\n" +
                "                    \"name\": \"张三\"\n" +
                "                 }\n" +
                "        }";
        JSONObject jsonValue = (JSONObject) JSON.parse(s);
        initUser(jsonValue);
    }



}
